package com.wuwangfu.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Arrays;
import java.util.List;

/**
 * @Author jcshen
 * @Date 2023-02-21
 * @PackageName:com.wuwangfu.source
 * @ClassName: CustomSourceA
 * @Description: 自定义Source
 * @Version 1.0
 */
public class CustomSourceA {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(2);

        DataStreamSource<String> line = env.addSource(new MySource());

        line.print();
        env.execute();
    }

    /**
     * 实现SourceFunction接口的Source，并行度为1，即单并行度
     */
    private static class MySource implements SourceFunction<String> {

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            List<String> words = Arrays.asList("aaa", "bbb", "ccc", "ddd", "eee");

            for (String word:words){
                //将Source产生的数据输出
                ctx.collect(word);
            }
        }

        @Override
        public void cancel() {

        }
    }
}
